package org.ws.task.schedule.dispatcher;

import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.model.NodeMetadata;
import org.ws.task.model.NodeStatus;
import org.ws.task.model.TaskConfig;
import org.ws.task.model.TaskItem;

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

public class PollDispatcher extends AbstractDispatcher{

    private static Logger log = LoggerFactory.getLogger(PollDispatcher.class.getName());

    private final AtomicInteger indexCount = new AtomicInteger(0);

    @Override
    public TaskConfig.DispatchStrategy strategy() {
        return TaskConfig.DispatchStrategy.POLL;
    }

    @Override
    protected List<NodeMetadata> getNode(TaskItem item,List<NodeMetadata> taskWorkerNodes) {
        final int size = taskWorkerNodes.size();
        int error = 0;
        NodeMetadata workerNode = null;
        int index = 0;
        do {
            index = indexCount.incrementAndGet() % size;
            workerNode = taskWorkerNodes.get(index);
            error++;
        }
        while (workerNode.getStatus() != NodeStatus.ACTIVE && error < size);
        log.debug("size:{},indexCount:{},index:{}",size,indexCount.get(),indexCount.get() % size);
        return Lists.newArrayList(workerNode);
    }

}
